package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * sku粒度，下单聚合统计
 *  维度：sku
 *  度量：原始金额、优惠券减免金额、活动减免金额、实付金额
 *  数据来源：dwd下单事实表
 * 开发流程
 *      从dwd下单事实表中读取数据
 *      对流中数据进行类型转换并过滤空消息 jsonStr->jsonObj
 *      去重
 *          为什么要进行去重操作？
 *              因为dwd下单事实表由4张表组成，订单明细、订单、订单明细活动、订单明细优惠券
 *              上述4张表在进行关联的时候，订单明细是主表
 *              和订单表进行关联的时候使用的是内连接
 *              和订单明细活动、订单明细优惠券使用的是左外连接
 *              左外连接，如果左表数据先到，右表数据后到，会产生3条数据
 *                  左表  null    +I
 *                  左表  null    -D
 *                  左表  右表     +I
 *              这样的结果发送到kafka主题，kafka主题会接收到3条消息
 *                  左表  null
 *                  null
 *                  左表  右表
 *              我们要从这样的kafka主题中读取数据，需要过滤空消息以及去重
 *          按照唯一键订单明细id进行分组
 *          去重方式1：状态 + 定时器      优点：如果有重复，只会向下游传递一条数据，不会出现膨胀  缺点：影响时效性
 *              当第1条数据进来的时候，先放到状态中，并注册5s后执行的定时器
 *              当第2条重复数据进来的时候，会用当前这条数据的聚合时间和状态中的数据的聚合时间进行比较，将时间大放到状态中
 *              等到定时器触发计算的时候，将状态的数据向下游传递
 *          去重方式2：状态 + 抵消       优点：时效性好     缺点：如果出现重复，会有数据膨胀的情况
 *              当第1条数据进来的时候，先放到状态中，并直接传递到下游
 *              当第2条重复数据进来的时候，先将状态中影响到度量值的字段进行取反，传递到下游
 *              在将第2条数据本身传递到下游
 *
 *      再次对流中数据进行转换 jsonObj->实体类对象
 *      指定Watermark的生成策略以及提取事件时间字段
 *      按照统计的维度sku进行分组
 *      开窗
 *      聚合计算
 *      维度关联
 *          版本1：最基本的维度关联
 *              在HBaseUtil--->getRow
 *          版本2：旁路缓存
 *              思路：先从缓存中获取维度数据，如果从缓存中获取到了要关联的维度，直接将其作为方法的返回值进行返回(缓存命中)
 *                   如果在缓存中没有获取到对应的维度数据，发送请求到HBase中查询维度，并将查询结果放到缓存中缓存起来，方便下次查询使用
 *              选型：
 *                  状态：     性能更好        维护性差
 *                  Redis     性能也不错      维护性好   √
 *              关于Redis的配置
 *                  key：    维度表名:主键值  例如：   dim_base_trademark:1
 *                  type:   String
 *                  expire: 1day    避免冷数据常驻内存，给内存带来压力
 *                  注意：如果维度数据发生了变化，需要将缓存数据清除掉
 *                          DimSinkFunction->清缓存
 *              RedisUtil
 *                  getJedis、closeJedis、readDim、writeDim
 *          版本3：旁路缓存 + 模板方法
 *              DimMapFunction
 *          版本4：旁路缓存 + 异步IO
 *              为什么使用异步IO？
 *                  如果要向提升程序的处理能力，可以调高算子的并行度，但是更大的并行度意味着需要更多的硬件资源，不可能无限制的调高
 *                  在资源有限的情况下，在单个并行度上，可以通过异步的方式提升处理能力
 *                  默认：如果使用map->MapFunction对流中数据进行处理，使用的是同步的方式
 *              Flink提供了专门发送异步请求的API
 *                  AsyncDataStream.[un]orderedWait(
 *                      流,
 *                      如果发送异步请求--实现AsyncFunction接口,
 *                      超时时间,
 *                      时间单位
 *                  )
 *              如果要向通过异步的方式进行维度关联，除了Flink之外，还需要让Redis和HBase提供异步操作的方式
 *                  RedisUtil->StatefulRedisConnection->readDimAsync|writeDimAsync
 *                  HBaseUtil->AsyncConnection->readDimAsync
 *          版本5： 旁路缓存 + 异步IO + 模板方法
 *              DimAsyncFunction->asyncInvoke
 *                  //创建异步编排对象，有返回值
 *                  CompletableFuture
 *                      .supplyAsync        当前线程任务返回值将作为下一个线程任务入参
 *                      .thenApplyAsync     //执行线程任务  有入参、有返回值
 *                      .thenAcceptAsync    //执行线程任务   有入参、无返回值
 *      将关联结果写到Doris中
 *
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、doris、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 *
 */
public class DwsTradeSkuOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeSkuOrderWindow().start(
                10029,
                4,
                "dws_trade_sku_order_window",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );

    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 1.对流中数据进行类型转换 并过滤空消息
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            out.collect(JSON.parseObject(jsonStr));
                        }
                    }
                }
        );
        //jsonObjDS.print();
        //TODO 2.按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //TODO 3.去重
        /*//3.1 去重方案1：状态 + 定时器 优点：如果出现重复，也只会向下游发送一条消息，没有数据膨胀  缺点：不管数据是否重复，都要等5s才会传递到下游，时效性差
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        //从状态中获取上条数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj == null) {
                            //说明状态中没有上条数据，属于没有重复的情况，将当前数据放到状态中
                            lastJsonObjState.update(jsonObj);
                            //并注册5s后执行的定时器
                            TimerService timerService = ctx.timerService();
                            long currentProcessingTime = timerService.currentProcessingTime();
                            timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        } else {
                            //说明状态中有上条数据，属于重复的情况，用当前数据的聚合时间和状态中数据的聚合时间进行比较，将时间大的放到状态中
                            String ts1 = lastJsonObj.getString("聚合时间");
                            String ts2 = jsonObj.getString("聚合时间");
                            if (ts2.compareTo(ts1) >= 0) {
                                //将当前这条数据放到状态中
                                lastJsonObjState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        //当定时器触发的时候执行的方法
                        //获取状态中的数据
                        JSONObject jsonObj = lastJsonObjState.value();
                        //将数据向下游传递
                        out.collect(jsonObj);
                        //清状态
                        lastJsonObjState.clear();
                    }
                }
        );*/
        //3.2 去重方式2：状态 + 抵消  优点：时效性好    缺点：如果出现重复，会向下游发送3条消息，数据膨胀了
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(5)).build());
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        //从状态中获取上条数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if(lastJsonObj != null){
                            //说明重复了，将状态中影响到度量值的字段进行取反，再向下游传递
                            String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                            String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                            String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                            String splitTotalAmount = lastJsonObj.getString("split_total_amount");
                            lastJsonObj.put("split_original_amount","-" + splitOriginalAmount);
                            lastJsonObj.put("split_coupon_amount","-" + splitCouponAmount);
                            lastJsonObj.put("split_activity_amount","-" +  splitActivityAmount);
                            lastJsonObj.put("split_total_amount","-" + splitTotalAmount);
                            out.collect(lastJsonObj);
                        }
                        lastJsonObjState.update(jsonObj);
                        out.collect(jsonObj);
                    }
                }
        );

//        distinctDS.print();

        //TODO 4.再次对流中数据进行转换    jsonObj->实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = distinctDS.map(
                new MapFunction<JSONObject, TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                        //        {"create_time":"2024-09-04 10:16:22","sku_num":"1","activity_rule_id":"1","split_original_amount":"6999.0000",
                        //        "split_coupon_amount":"0.0","sku_id":"2","date_id":"2024-09-04","user_id":"2598","province_id":"1","activity_id":"1",
                        //        "sku_name":"小米12S Ultra 骁龙8+旗舰处理器 徕卡光学镜头 2K超视感屏 120Hz高刷 67W快充 12GB+512GB 冷杉绿 5G手机","id":"13960",
                        //        "order_id":"9938","split_activity_amount":"500.0","split_total_amount":"6499.0","ts":1725588982}
                        String skuId = jsonObj.getString("sku_id");
                        BigDecimal splitOriginalAmount = jsonObj.getBigDecimal("split_original_amount");
                        BigDecimal splitCouponAmount = jsonObj.getBigDecimal("split_coupon_amount");
                        BigDecimal splitActivityAmount = jsonObj.getBigDecimal("split_activity_amount");
                        BigDecimal splitTotalAmount = jsonObj.getBigDecimal("split_total_amount");
                        Long ts = jsonObj.getLong("ts") * 1000;
                        TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                                .skuId(skuId)
                                .originalAmount(splitOriginalAmount)
                                .couponReduceAmount(splitCouponAmount)
                                .activityReduceAmount(splitActivityAmount)
                                .orderAmount(splitTotalAmount)
                                .ts(ts)
                                .build();
                        return orderBean;
                    }
                }
        );
//        orderBeanDS.print();

        //TODO 5.指定Watermark生成策略以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeSkuOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                                        return orderBean.getTs();
                                    }
                                }
                        )
        );

        //TODO 6.按照统计的维度sku进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 7.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 8.聚合计算
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setCouponReduceAmount(value1.getCouponReduceAmount().add(value2.getCouponReduceAmount()));
                        value1.setActivityReduceAmount(value1.getActivityReduceAmount().add(value2.getActivityReduceAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                },
                new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean orderBean = input.iterator().next();
                        String stt = DateFormatUtil.tsToDateTime(window.getStart());
                        String edt = DateFormatUtil.tsToDateTime(window.getEnd());
                        String curDate = DateFormatUtil.tsToDate(window.getStart());
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setCurDate(curDate);
                        out.collect(orderBean);
                    }
                }
        );
        //reduceDS.print();

        //TODO 9.关联sku维度
        /*
        //维度关联最基本实现
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private Connection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                        //根据流中的对象获取要关联的维度主键
                        String skuId = orderBean.getSkuId();
                        //根据维度主键获取对应的维度对象
                        //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        JSONObject dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);
                        //将维度对象属性补充到流中对象上
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        return orderBean;
                    }
                }
        );
        withSkuInfoDS.print();

        //优化1：旁路缓存
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private Connection hbaseConn;
                    private Jedis jedis;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHBaseConnection();
                        jedis = RedisUtil.getJedis();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                        RedisUtil.closeJedis(jedis);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                        //根据当前流中的对象获取要关联的维度主键
                        String skuId = orderBean.getSkuId();
                        //先根据主键到Redis中获取维度数据
                        JSONObject dimJsonObj = RedisUtil.readDim(jedis, "dim_sku_info", skuId);
                        if (dimJsonObj != null) {
                            //如果在Redis中获取到了维度数据---直接将其返回（缓存命中）
                            System.out.println("~~从Redis中获取维度数据~~~");
                        } else {
                            //如果在Redis中没有获取到对应的维度数据，发送请求到Hbase中查询维度
                            dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);
                            if (dimJsonObj != null) {
                                System.out.println("~~从HBase中获取维度数据~~~");
                                //并将从HBase中查询出来的维度放到Redis中缓存起来
                                RedisUtil.writeDim(jedis, "dim_sku_info", skuId, dimJsonObj);
                            } else {
                                System.out.println("~~~没有找到要关联的维度~~~");
                            }

                        }
                        //将维度属性补充到流中对象上
                        if (dimJsonObj != null) {
                            orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                            orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                            orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                            orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        }
                        return orderBean;
                    }
                }
        );
        //withSkuInfoDS.print();

        //优化1：旁路缓存 + 模板方法设计模式
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new DimMapFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_sku_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSkuId();
                    }
                }
        );

        withSkuInfoDS.print();


        //优化2： 旁路缓存  + 异步IO
        //将异步I/O操作应用于DataStream作为 DataStream 的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                //如何发送异步请求----需要实现AsyncFunction
                new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private StatefulRedisConnection<String, String> asyncRedisConn;
                    private AsyncConnection asyncHBaseConn;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        asyncRedisConn = RedisUtil.getRedisAsyncConnection();
                        asyncHBaseConn = HBaseUtil.getHBaseAsyncConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        RedisUtil.closeRedisAsyncConnection(asyncRedisConn);
                        HBaseUtil.closeHBaseAsyncConnection(asyncHBaseConn);
                    }

                    @Override
                    public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                        //根据流中对象获取要关联的维度的主键
                        String skuId = orderBean.getSkuId();
                        //先以异步的方式从Redis中获取要关联的维度
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(asyncRedisConn, "dim_sku_info", skuId);
                        if(dimJsonObj != null){
                            //如果从Redis中查询到了要关联的维度，直接将其作为返回值进行返回(缓存命中)
                            System.out.println("~~~从Redis中获取维度数据~~~");
                        }else {
                            //如果从Redis中没有查询到要关联的维度，发送异步请求到HBase中查询维度
                            dimJsonObj = HBaseUtil.readDimAsync(asyncHBaseConn,Constant.HBASE_NAMESPACE,"dim_sku_info",skuId);
                            if(dimJsonObj != null){
                                //将从HBase中查询出来的维度以异步的方式放到Redis中缓存起来
                                System.out.println("~~~从HBase中获取维度数据~~~");
                                RedisUtil.writeDimAsync(asyncRedisConn,"dim_sku_info", skuId,dimJsonObj);
                            }else{
                                System.out.println("~~~没有找到要关联的维度数据~~~");
                            }
                        }
                        //将维度对象相关的属性补充到流中对象上
                        if(dimJsonObj != null){
                            orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                            orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                            orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                            orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        }
                        //获取数据库交互的结果并发送给 ResultFuture 的 回调 函数--向下游传递数据
                        resultFuture.complete(Collections.singleton(orderBean));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //withSkuInfoDS.print();

         */

        //优化2： 旁路缓存  + 异步IO  + 模板方法
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_sku_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSkuId();
                    }
                },
                60, TimeUnit.SECONDS
        );

        //withSkuInfoDS.print();

        //TODO 10.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
                withSkuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSpuName(dimJsonObj.getString("spu_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_spu_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSpuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //TODO 11.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
                withSpuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setTrademarkName(dimJsonObj.getString("tm_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_trademark";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getTrademarkId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //TODO 12.关联Category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
                withTmDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category3";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory3Name(dim.getString("name"));
                        bean.setCategory2Id(dim.getString("category2_id"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        //TODO 13.关联Category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
                c3Stream,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category2";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory2Name(dim.getString("name"));
                        bean.setCategory1Id(dim.getString("category1_id"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        //TODO 14.关联Category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withC1DS = AsyncDataStream.unorderedWait(
                c2Stream,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category1";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory1Name(dim.getString("name"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        withC1DS.print();

        //TODO 15.将关联结果写到Doris
        withC1DS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_sku_order_window"));

    }
}
